package com.my.study.rocketmq.batch;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * 批量-生产者
 *
 * @author yyb8152
 * @since 20220510
 **/
public class Producer {
    public static void main(String[] args) throws Exception {
        //1.创建消息生产者producer，并制定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        //2.指定Nameserver地址
        producer.setNamesrvAddr("39.101.165.196:9876");
        //3.启动producer
        producer.start();

        // 创建message集合对象
        List<Message> messageList = new ArrayList<>();

        //4.创建消息对象，指定主题Topic、Tag和消息体
        /**
         * 参数一：消息主题Topic
         * 参数二：消息Tag
         * 参数三：消息内容
         */

        messageList.add(new Message("BatchTopic", "Tag1", "Hello rocketMQ1".getBytes()));
        messageList.add(new Message("BatchTopic", "Tag2", "Hello rocketMQ2".getBytes()));
        messageList.add(new Message("BatchTopic", "Tag3", "Hello rocketMQ3".getBytes()));
        messageList.add(new Message("BatchTopic", "Tag4", "Hello rocketMQ4".getBytes()));


        // 优化如果消息超过4M需要进行分割
        //把大的消息分裂成若干个小的消息
        ListSplitter splitter = new ListSplitter(messageList);
        while (splitter.hasNext()) {
            try {
                List<Message> listItem = splitter.next();
                //5.发送同步消息
                SendResult result = producer.send(listItem);
                System.out.println("发送结果：" + result);
            } catch (Exception e) {
                e.printStackTrace();
                //处理error
            }
        }

        // 休眠1s
        TimeUnit.SECONDS.sleep(1);


        //6.关闭生产者producer
        producer.shutdown();
    }


}
